package cn.data.stream.stream;

import cn.data.stream.expection.DataStreamInitException;
import cn.data.stream.expection.DataStreamProcessException;
import com.google.common.base.Throwables;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.env.PropertyResolver;

/**
 * @author shi hong yu
 */
@Slf4j
public abstract class AbstractStream implements IComponent{

    protected String name;

    /**
     * spring 配置文件 可读取本stream所需配置
     */
    @Resource
    protected PropertyResolver config;

    @Override
    public void init() throws DataStreamInitException {

    }

    public void setName(String name){
        this.name = name;
    }

    @Override
    public String getName() {
        return name;
    }

    private List<IComponent> outputs= new ArrayList<>();

    @Override
    public void addOutputStream(List<IComponent> out) {
        if(out!=null){
            out.stream().filter(Objects::nonNull).forEach(o->outputs.add(o));
        }
    }

    @Override
    public void addOutputStream(IComponent out) {
        if(out!=null){
            outputs.add(out);
        }
    }

    @Override
    public List<IComponent> getOutputStream() {
        return Collections.unmodifiableList(outputs);
    }

    @Override
    public void process(Object data) {
        if(outputs!=null){
            for(IComponent out:outputs){
                log.debug("数据发往下游：{}->{}",this.name,out.getName());
                try {
                    out.process(data);
                } catch (Exception e) {
                    log.warn("数据发往下游：{}->{} 失败",this.name,out.getName(), Throwables.getStackTraceAsString(e));
                }
            }
        }else{
            log.warn("{} 没有下游组件 数据不下发",this.name);
        }
    }

    @Override
    public void destroy() {

    }

}
